#include <srs_app_dvr.hpp>

#ifdef SRS_AUTO_DVR

#include <fcntl.h>
#include <sstream>
#include <algorithm>
#include <thread>
using namespace std;

#include <srs_app_config.hpp>
#include <srs_rtmp_stack.hpp>
#include <srs_core_autofree.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_app_http_hooks.hpp>
#include <srs_kernel_codec.hpp>
#include <srs_kernel_flv.hpp>
#include <srs_kernel_file.hpp>
#include <srs_rtmp_amf0.hpp>
#include <srs_kernel_stream.hpp>
#include <srs_protocol_json.hpp>
#include <srs_app_utility.hpp>

#include <utils/tools.h>
#include <utils/tools_curl.h>

// update the flv duration and filesize every this interval in ms.
#define SRS_DVR_UPDATE_DURATION_INTERVAL 60000

SrsFlvSegment::SrsFlvSegment(SrsDvrPlan* p) {
    req = nullptr;
    jitter = nullptr;
    plan = p;

    fs = new SrsFileWriter();
    enc = new SrsFlvEncoder();
    jitter_algorithm = SrsRtmpJitterAlgorithmOFF;

    path = "";
    has_keyframe = false;
    duration = 0;
    starttime = -1;
    stream_starttime = 0;
    stream_previous_pkt_time = -1;
    stream_duration = 0;

    duration_offset = 0;
    filesize_offset = 0;

    _srs_config->subscribe(this);
}

SrsFlvSegment::~SrsFlvSegment() {
    _srs_config->unsubscribe(this);

    srs_freep(jitter);
    srs_freep(fs);
    srs_freep(enc);
}

int SrsFlvSegment::initialize(SrsRequest* r) {
    int ret = ERROR_SUCCESS;
    req = r;
    jitter_algorithm = (SrsRtmpJitterAlgorithm)_srs_config->get_dvr_time_jitter(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));
    return ret;
}

bool SrsFlvSegment::is_overflow(int64_t max_duration) {
    return duration >= max_duration;
}

int SrsFlvSegment::open(bool use_tmp_file) {
    int ret = ERROR_SUCCESS;

    // ignore when already open.
    if (fs->is_open()) {
        return ret;
    }

    path = generate_path();
    bool fresh_flv_file = !srs_path_exists(path);

    // create dir first.
    string dir = path.substr(0, path.rfind("/"));
    if ((ret = srs_create_dir_recursively(dir)) != ERROR_SUCCESS) {
        srs_error("create dir=%s failed. ret=%d", dir.c_str(), ret);
        return ret;
    }
    srs_info("create dir=%s ok", dir.c_str());

    // create jitter.
    if ((ret = create_jitter(!fresh_flv_file)) != ERROR_SUCCESS) {
        srs_error("create jitter failed, path=%s, fresh=%d. ret=%d", path.c_str(), fresh_flv_file, ret);
        return ret;
    }

    // generate the tmp flv path.
    if (!fresh_flv_file || !use_tmp_file) {
        // when path exists, always append to it.
        // so we must use the target flv path as output flv.
        tmp_flv_file = path;
    } else {
        // when path not exists, dvr to tmp file.
        tmp_flv_file = path + ".tmp";
    }

    // open file writer, in append or create mode.
    if (!fresh_flv_file) {
        if ((ret = fs->open_append(tmp_flv_file)) != ERROR_SUCCESS) {
            srs_error("append file stream for file %s failed. ret=%d", path.c_str(), ret);
            return ret;
        }
        srs_trace("dvr: always append to when exists, file=%s.", path.c_str());
    } else {
        if ((ret = fs->open(tmp_flv_file)) != ERROR_SUCCESS) {
            srs_error("open file stream for file %s failed. ret=%d", path.c_str(), ret);
            return ret;
        }
    }

    // initialize the encoder.
    if ((ret = enc->initialize(fs)) != ERROR_SUCCESS) {
        srs_error("initialize enc by fs for file %s failed. ret=%d", path.c_str(), ret);
        return ret;
    }

    // when exists, donot write flv header.
    if (fresh_flv_file) {
        // write the flv header to writer.
        if ((ret = enc->write_header()) != ERROR_SUCCESS) {
            srs_error("write flv header failed. ret=%d", ret);
            return ret;
        }
    }

    // update the duration and filesize offset.
    duration_offset = 0;
    filesize_offset = 0;

    srs_trace("dvr stream %s to file %s", req->stream.c_str(), path.c_str());
    return ret;
}

int SrsFlvSegment::close(SrsRequest* req) {
    int ret = ERROR_SUCCESS;

    // ignore when already closed.
    if (!fs->is_open()) {
        return ret;
    }

    // update duration and filesize.
    if ((ret = update_flv_metadata()) != ERROR_SUCCESS) {
        return ret;
    }

    fs->close();

    // when tmp flv file exists, reap it.
    if (tmp_flv_file != path) {
        if (rename(tmp_flv_file.c_str(), path.c_str()) < 0) {
            ret = ERROR_SYSTEM_FILE_RENAME;
            srs_error("rename flv file failed, %s => %s. ret=%d", tmp_flv_file.c_str(), path.c_str(), ret);
            return ret;
        }
    }

    uploadDvrFiles(req, path);

    // TODO: FIXME: the http callback is async, which will trigger thread switch,
    //          so the on_video maybe invoked during the http callback, and error.
    if ((ret = plan->on_reap_segment(req)) != ERROR_SUCCESS) {
        srs_error("dvr: notify plan to reap segment failed. ret=%d", ret);
        return ret;
    }
    return ret;
}

void SrsFlvSegment::uploadDvrFiles(SrsRequest* req, const std::string& file) {
    if (req == nullptr) {
        return;
    }

    auto pf = randSrsDvrPostFile();
    if (pf == nullptr) {
        return;
    }

    auto cfgDvr = _srs_config->get_dvr(req->vhost);
    auto argDvr = req->get_args_obj("dvr");

    auto& dvrStore = _srs_config->get_dvr_store(cfgDvr, argDvr);
    if (dvrStore.empty() == true) {
        return;
    }

    auto& dvrPath = _srs_config->get_dvr_path(cfgDvr, argDvr);
    if (dvrPath.empty() == true) {
        return;
    }

    string fsuffix, fpath;
    auto pos = dvrPath.rfind(".");
    if (pos != string::npos) {
        fsuffix = dvrPath.substr(pos + 1);
    }
    pos = dvrPath.rfind("/");
    if (pos != string::npos) {
        fpath = dvrPath.substr(0, pos);
    }

    if (fsuffix.empty() == true || fpath.empty() == true) {
        return;
    }

    fpath = srs_path_build_stream(fpath, req->vhost, req->app, req->stream);
    fpath = srs_path_build_timestamp(fpath);

    if (srs_path_exists(fpath) == false) {
        srs_error("invalid dvr path=%s, vhost=%s", fpath.c_str(), req->vhost.c_str());
        return;
    }

    vector<string> files;
    if (file.empty() == false) {
        files.push_back(file);
    } else {
        files = srs_file_search(fpath, fsuffix);
    }
    if (files.empty() == true) {
        return;
    }

    pf->add(dvrStore, files);
}

int SrsFlvSegment::write_metadata(SrsSharedPtrMessage* metadata) {
    int ret = ERROR_SUCCESS;
    if (duration_offset || filesize_offset) {
        return ret;
    }

    SrsStream stream;
    if ((ret = stream.initialize(metadata->payload, metadata->size)) != ERROR_SUCCESS) {
        return ret;
    }

    SrsAmf0Any* name = SrsAmf0Any::str();
    SrsAutoFree(SrsAmf0Any, name);
    if ((ret = name->read(&stream)) != ERROR_SUCCESS) {
        return ret;
    }

    SrsAmf0Object* obj = SrsAmf0Any::object();
    SrsAutoFree(SrsAmf0Object, obj);
    if ((ret = obj->read(&stream)) != ERROR_SUCCESS) {
        return ret;
    }

    // remove duration and filesize.
    obj->set("filesize", nullptr);
    obj->set("duration", nullptr);

    // add properties.
    obj->set("service", SrsAmf0Any::str(RTMP_SIG_SRS_SERVER));
    obj->set("filesize", SrsAmf0Any::number(0));
    obj->set("duration", SrsAmf0Any::number(0));

    int size = name->total_size() + obj->total_size();
    char* payload = new char[size];
    SrsAutoFreeA(char, payload);

    // 11B flv header, 3B object EOF, 8B number value, 1B number flag.
    duration_offset = fs->tellg() + size + 11 - SrsAmf0Size::object_eof() - SrsAmf0Size::number();
    // 2B string flag, 8B number value, 8B string 'duration', 1B number flag
    filesize_offset = duration_offset - SrsAmf0Size::utf8("duration") - SrsAmf0Size::number();

    // convert metadata to bytes.
    if ((ret = stream.initialize(payload, size)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = name->write(&stream)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = obj->write(&stream)) != ERROR_SUCCESS) {
        return ret;
    }

    // to flv file.
    if ((ret = enc->write_metadata(18, payload, size)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsFlvSegment::write_audio(SrsSharedPtrMessage* shared_audio) {
    int ret = ERROR_SUCCESS;

    SrsSharedPtrMessage* audio = shared_audio->copy();
    SrsAutoFree(SrsSharedPtrMessage, audio);

    if ((jitter->correct(audio, jitter_algorithm)) != ERROR_SUCCESS) {
        return ret;
    }

    char* payload = audio->payload;
    int size = audio->size;
    int64_t timestamp = plan->filter_timestamp(audio->timestamp);
    if ((ret = enc->write_audio(timestamp, payload, size)) != ERROR_SUCCESS) {
        return ret;
    }

    if ((ret = on_update_duration(audio)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsFlvSegment::write_video(SrsSharedPtrMessage* shared_video) {
    int ret = ERROR_SUCCESS;

    SrsSharedPtrMessage* video = shared_video->copy();
    SrsAutoFree(SrsSharedPtrMessage, video);

    char* payload = video->payload;
    int size = video->size;

    bool is_sequence_header = SrsFlvCodec::video_is_sequence_header(payload, size);
#ifdef SRS_AUTO_HTTP_CALLBACK
    bool is_key_frame = SrsFlvCodec::video_is_h264(payload, size)
                        && SrsFlvCodec::video_is_keyframe(payload, size) && !is_sequence_header;
    if (is_key_frame) {
        has_keyframe = true;
        if ((ret = plan->on_video_keyframe()) != ERROR_SUCCESS) {
            return ret;
        }
    }
    srs_verbose("dvr video is key: %d", is_key_frame);
#endif

    // accept the sequence header here.
    // when got no keyframe, ignore when should wait keyframe.
    if (!has_keyframe && !is_sequence_header) {
        bool wait_keyframe = _srs_config->get_dvr_wait_keyframe(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));
        if (wait_keyframe) {
            srs_info("dvr: ignore when wait keyframe.");
            return ret;
        }
    }

    if ((jitter->correct(video, jitter_algorithm)) != ERROR_SUCCESS) {
        return ret;
    }

    // update segment duration, session plan just update the duration,
    // the segment plan will reap segment if exceed, this video will write to next segment.
    if ((ret = on_update_duration(video)) != ERROR_SUCCESS) {
        return ret;
    }

    int32_t timestamp = (int32_t)plan->filter_timestamp(video->timestamp);
    if ((ret = enc->write_video(timestamp, payload, size)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsFlvSegment::update_flv_metadata() {
    int ret = ERROR_SUCCESS;

    // no duration or filesize specified.
    if (!duration_offset || !filesize_offset) {
        return ret;
    }

    int64_t cur = fs->tellg();

    // buffer to write the size.
    char* buf = new char[SrsAmf0Size::number()];
    SrsAutoFreeA(char, buf);

    SrsStream stream;
    if ((ret = stream.initialize(buf, SrsAmf0Size::number())) != ERROR_SUCCESS) {
        return ret;
    }

    // filesize to buf.
    SrsAmf0Any* size = SrsAmf0Any::number((double)cur);
    SrsAutoFree(SrsAmf0Any, size);

    stream.skip(-1 * stream.pos());
    if ((ret = size->write(&stream)) != ERROR_SUCCESS) {
        return ret;
    }

    // update the flesize.
    fs->lseek(filesize_offset);
    if ((ret = fs->write(buf, SrsAmf0Size::number(), nullptr)) != ERROR_SUCCESS) {
        return ret;
    }

    // duration to buf
    SrsAmf0Any* dur = SrsAmf0Any::number((double)duration / 1000.0);
    SrsAutoFree(SrsAmf0Any, dur);

    stream.skip(-1 * stream.pos());
    if ((ret = dur->write(&stream)) != ERROR_SUCCESS) {
        return ret;
    }

    // update the duration
    fs->lseek(duration_offset);
    if ((ret = fs->write(buf, SrsAmf0Size::number(), nullptr)) != ERROR_SUCCESS) {
        return ret;
    }

    // reset the offset.
    fs->lseek(cur);

    return ret;
}

const string& SrsFlvSegment::get_path() {
    return path;
}

string SrsFlvSegment::generate_path() {
    // the path in config, for example,
    //      /data/[vhost]/[app]/[stream]/[2006]/[01]/[02]/[15].[04].[05].[999].flv
    auto path_config = _srs_config->get_dvr_path(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));

    // add [stream].[timestamp].flv as filename for dir
    if (path_config.find(".flv") != path_config.length() - 4) {
        path_config += string("/[stream]/[timestamp].flv");
    }

    // the flv file path
    path_config = srs_path_build_stream(path_config, req->vhost, req->app, req->stream);
    path_config = srs_path_build_timestamp(path_config);

    return path_config;
}

int SrsFlvSegment::create_jitter(bool loads_from_flv) {
    int ret = ERROR_SUCCESS;

    // when path exists, use exists jitter.
    if (!loads_from_flv) {
        // jitter when publish, ensure whole stream start from 0.
        srs_freep(jitter);
        jitter = new SrsRtmpJitter();

        // fresh stream starting.
        starttime = -1;
        stream_previous_pkt_time = -1;
        stream_starttime = srs_update_system_time_ms();
        stream_duration = 0;

        // fresh segment starting.
        has_keyframe = false;
        duration = 0;

        return ret;
    }

    // when jitter ok, do nothing.
    if (jitter) {
        return ret;
    }

    // always ensure the jitter crote.
    // for the first time, initialize jitter from exists file.
    jitter = new SrsRtmpJitter();

    // TODO: FIXME: implements it.
    return ret;
}

int SrsFlvSegment::on_update_duration(SrsSharedPtrMessage* msg) {
    int ret = ERROR_SUCCESS;

    // we must assumpt that the stream timestamp is monotonically increase,
    // that is, always use time jitter to correct the timestamp.
    // except the time jitter is disabled in config.

    // set the segment starttime at first time
    if (starttime < 0) {
        starttime = msg->timestamp;
    }

    // no previous packet or timestamp overflow.
    if (stream_previous_pkt_time < 0 || stream_previous_pkt_time > msg->timestamp) {
        stream_previous_pkt_time = msg->timestamp;
    }

    // collect segment and stream duration, timestamp overflow is ok.
    duration += msg->timestamp - stream_previous_pkt_time;
    stream_duration += msg->timestamp - stream_previous_pkt_time;

    // update previous packet time
    stream_previous_pkt_time = msg->timestamp;
    return ret;
}

int SrsFlvSegment::on_reload_vhost_dvr(const string& vhost) {
    int ret = ERROR_SUCCESS;
    jitter_algorithm = (SrsRtmpJitterAlgorithm)_srs_config->get_dvr_time_jitter(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));
    return ret;
}

SrsDvrAsyncCallOnDvr::SrsDvrAsyncCallOnDvr(int c, SrsRequest* r, const string& p) {
    cid = c;
    req = r->copy();
    path = p;
}

SrsDvrAsyncCallOnDvr::~SrsDvrAsyncCallOnDvr() {
    srs_freep(req);
}

int SrsDvrAsyncCallOnDvr::call() {
    int ret = ERROR_SUCCESS;

#ifdef SRS_AUTO_HTTP_CALLBACK
    auto phook = _srs_config->get_vhost_http_hooks(req->vhost);
    if (phook == nullptr) {
        return ret;
    }

    if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost, phook)) {
        return ret;
    }

    // the http hooks will cause context switch,
    // so we must copy all hooks for the on_connect may freed.
    auto hooks = _srs_config->get_vhost_on_dvr(req->vhost, phook);
    if (hooks.empty() == true) {
        srs_info("ignore the empty http callback: on_dvr");
        return ret;
    }

    for (size_t i = 0; i < hooks.size(); i++) {
        auto& url = hooks.at(i);
        if ((ret = SrsHttpHooks::on_dvr(cid, url, req, path)) != ERROR_SUCCESS) {
            srs_error("hook client on_dvr failed. url=%s, ret=%d", url.c_str(), ret);
            return ret;
        }
    }
#endif
    return ret;
}

string SrsDvrAsyncCallOnDvr::to_string() {
    std::stringstream ss;
    ss << "vhost=" << req->vhost << ", file=" << path;
    return ss.str();
}

SrsDvrPlan::SrsDvrPlan() {
    _req = nullptr;
    dvr_enabled = false;
    segment = new SrsFlvSegment(this);
    async = new SrsAsyncCallWorker();
}

SrsDvrPlan::~SrsDvrPlan() {
    srs_freep(segment);
    srs_freep(async);
}

int SrsDvrPlan::initialize(SrsRequest* r) {
    int ret = ERROR_SUCCESS;
    _req = r;

    if ((ret = segment->initialize(r)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = async->start()) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrPlan::on_video_keyframe() {
    return ERROR_SUCCESS;
}

int64_t SrsDvrPlan::filter_timestamp(int64_t timestamp) {
    return timestamp;
}

int SrsDvrPlan::on_meta_data(SrsSharedPtrMessage* shared_metadata) {
    int ret = ERROR_SUCCESS;
    if (!dvr_enabled) {
        return ret;
    }
    return segment->write_metadata(shared_metadata);
}

int SrsDvrPlan::on_audio(SrsSharedPtrMessage* shared_audio) {
    int ret = ERROR_SUCCESS;
    if (!dvr_enabled) {
        return ret;
    }
    if ((ret = segment->write_audio(shared_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrPlan::on_video(SrsSharedPtrMessage* shared_video) {
    int ret = ERROR_SUCCESS;
    if (!dvr_enabled) {
        return ret;
    }
    if ((ret = segment->write_video(shared_video)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrPlan::on_reap_segment(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    int ret = ERROR_SUCCESS;
    int cid = _srs_context->get_id();
    if ((ret = async->execute(new SrsDvrAsyncCallOnDvr(cid, req, segment->get_path()))) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

bool SrsDvrPlan::args_dvr_enabled(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    if (req == nullptr || req->args == nullptr) {
        return true;
    }
    SrsAmf0Object* cdvr = nullptr;
    if (req->args != nullptr) {
        auto cpro = req->args->get_property("dvr");
        if (cpro != nullptr) {
            cdvr = cpro->to_object();
        }
    }
    if (cdvr != nullptr && cdvr->sure_property_integer("id") <= 0) {
        return false;
    }
    return true;
}

SrsDvrPlan* SrsDvrPlan::create_plan(SrsRequest* req) {
    if (req != nullptr) {
        auto& plan = _srs_config->get_dvr_plan(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));
        if (srs_config_dvr_is_plan_segment(plan)) {
            return new SrsDvrSegmentPlan();
        } else if (srs_config_dvr_is_plan_session(plan)) {
            return new SrsDvrSessionPlan();
        } else if (srs_config_dvr_is_plan_append(plan)) {
            return new SrsDvrAppendPlan();
        } else {
            srs_error("invalid dvr plan=%s, vhost=%s", plan.c_str(), req->vhost.c_str());
            srs_assert(false);
        }
    }
    return nullptr;
}

SrsDvrSessionPlan::SrsDvrSessionPlan() {
}

SrsDvrSessionPlan::~SrsDvrSessionPlan() {
}

int SrsDvrSessionPlan::on_publish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    int ret = ERROR_SUCCESS;
    // support multiple publish.
    if (dvr_enabled) {
        return ret;
    }
    if (_srs_config->get_dvr_enabled(req->vhost) == false || args_dvr_enabled(req) == false) {
        return ret;
    }
    if ((ret = segment->close(req)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = segment->open()) != ERROR_SUCCESS) {
        return ret;
    }

    dvr_enabled = true;
    return ret;
}

void SrsDvrSessionPlan::on_unpublish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    // support multiple publish.
    if (!dvr_enabled) {
        return;
    }
    // ignore error.
    int ret = segment->close(req);
    if (ret != ERROR_SUCCESS) {
        srs_warn("ignore flv close error. ret=%d", ret);
    }
    dvr_enabled = false;
}

SrsDvrAppendPlan::SrsDvrAppendPlan() {
    last_update_time = 0;
}

SrsDvrAppendPlan::~SrsDvrAppendPlan() {
}

int SrsDvrAppendPlan::on_publish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    int ret = ERROR_SUCCESS;
    // support multiple publish.
    if (dvr_enabled) {
        return ret;
    }
    if (_srs_config->get_dvr_enabled(req->vhost) == false || args_dvr_enabled(req) == false) {
        return ret;
    }
    if ((ret = segment->open(false)) != ERROR_SUCCESS) {
        return ret;
    }
    dvr_enabled = true;
    return ret;
}

void SrsDvrAppendPlan::on_unpublish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
}

int SrsDvrAppendPlan::on_audio(SrsSharedPtrMessage* shared_audio) {
    int ret = ERROR_SUCCESS;
    if ((ret = update_duration(shared_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = SrsDvrPlan::on_audio(shared_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrAppendPlan::on_video(SrsSharedPtrMessage* shared_video) {
    int ret = ERROR_SUCCESS;
    if ((ret = update_duration(shared_video)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = SrsDvrPlan::on_video(shared_video)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrAppendPlan::update_duration(SrsSharedPtrMessage* msg) {
    int ret = ERROR_SUCCESS;
    if (last_update_time <= 0) {
        last_update_time = msg->timestamp;
        return ret;
    }

    if (msg->timestamp < last_update_time) {
        last_update_time = msg->timestamp;
        return ret;
    }

    if (SRS_DVR_UPDATE_DURATION_INTERVAL > msg->timestamp - last_update_time) {
        return ret;
    }
    last_update_time = msg->timestamp;

    srs_assert(segment);
    if (!segment->update_flv_metadata()) {
        return ret;
    }
    return ret;
}

SrsDvrSegmentPlan::SrsDvrSegmentPlan() {
    segment_duration = -1;
    metadata = sh_video = sh_audio = nullptr;
}

SrsDvrSegmentPlan::~SrsDvrSegmentPlan() {
    srs_freep(sh_video);
    srs_freep(sh_audio);
    srs_freep(metadata);
}

int SrsDvrSegmentPlan::initialize(SrsRequest* req) {
    int ret = ERROR_SUCCESS;
    if ((ret = SrsDvrPlan::initialize(req)) != ERROR_SUCCESS) {
        return ret;
    }

    segment_duration = _srs_config->get_dvr_duration(_srs_config->get_dvr(req->vhost), req->get_args_obj("dvr"));
    // to ms
    segment_duration *= 1000;
    return ret;
}

int SrsDvrSegmentPlan::on_publish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
    int ret = ERROR_SUCCESS;
    // support multiple publish.
    if (dvr_enabled) {
        return ret;
    }
    if (_srs_config->get_dvr_enabled(req->vhost) == false || args_dvr_enabled(req) == false) {
        return ret;
    }
    if ((ret = segment->close(req)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = segment->open()) != ERROR_SUCCESS) {
        return ret;
    }
    dvr_enabled = true;
    return ret;
}

void SrsDvrSegmentPlan::on_unpublish(SrsRequest* req) {
    if (req == nullptr) {
        req = _req;
    }
}

int SrsDvrSegmentPlan::on_meta_data(SrsSharedPtrMessage* shared_metadata) {
    int ret = ERROR_SUCCESS;

    srs_freep(metadata);
    metadata = shared_metadata->copy();

    if ((ret = SrsDvrPlan::on_meta_data(shared_metadata)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrSegmentPlan::on_audio(SrsSharedPtrMessage* shared_audio) {
    int ret = ERROR_SUCCESS;
    if (SrsFlvCodec::audio_is_sequence_header(shared_audio->payload, shared_audio->size)) {
        srs_freep(sh_audio);
        sh_audio = shared_audio->copy();
    }
    if ((ret = update_duration(shared_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = SrsDvrPlan::on_audio(shared_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrSegmentPlan::on_video(SrsSharedPtrMessage* shared_video) {
    int ret = ERROR_SUCCESS;
    if (SrsFlvCodec::video_is_sequence_header(shared_video->payload, shared_video->size)) {
        srs_freep(sh_video);
        sh_video = shared_video->copy();
    }
    if ((ret = update_duration(shared_video)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = SrsDvrPlan::on_video(shared_video)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvrSegmentPlan::update_duration(SrsSharedPtrMessage* msg) {
    if (segment == nullptr) {
        return ERROR_NO_SUITABLE_PTR;
    }

    int ret = ERROR_SUCCESS;
    // ignore if duration ok.
    if (_req == nullptr || segment_duration <= 0 || !segment->is_overflow(segment_duration)) {
        return ret;
    }

    // when wait keyframe, ignore if no frame arrived.
    if (_srs_config->get_dvr_wait_keyframe(_srs_config->get_dvr(_req->vhost), _req->get_args_obj("dvr"))) {
        if (!msg->is_video()) {
            return ret;
        }

        char* payload = msg->payload;
        int size = msg->size;
        bool is_key_frame = SrsFlvCodec::video_is_h264(payload, size)
                            && SrsFlvCodec::video_is_keyframe(payload, size)
                            && !SrsFlvCodec::video_is_sequence_header(payload, size);
        if (!is_key_frame) {
            return ret;
        }
    }

    // reap segment
    if ((ret = segment->close(_req)) != ERROR_SUCCESS) {
        return ret;
    }

    // open new flv file
    if ((ret = segment->open()) != ERROR_SUCCESS) {
        return ret;
    }

    // update sequence header
    if (metadata && (ret = SrsDvrPlan::on_meta_data(metadata)) != ERROR_SUCCESS) {
        return ret;
    }
    if (sh_video && (ret = SrsDvrPlan::on_video(sh_video)) != ERROR_SUCCESS) {
        return ret;
    }
    if (sh_audio && (ret = SrsDvrPlan::on_audio(sh_audio)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

SrsDvr::SrsDvr() {
    source = nullptr;
    plan = nullptr;
}

SrsDvr::~SrsDvr() {
    srs_freep(plan);
}

int SrsDvr::initialize(SrsSource* s, SrsRequest* r) {
    int ret = ERROR_SUCCESS;
    source = s;

    if (plan == nullptr) {
        plan = SrsDvrPlan::create_plan(r);
    }
    if ((ret = plan->initialize(r)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = source->on_dvr_request_sh()) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvr::on_publish(SrsRequest* r) {
    int ret = ERROR_SUCCESS;
    if ((ret = plan->on_publish(r)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

void SrsDvr::on_unpublish(SrsRequest* r) {
    plan->on_unpublish(r);
}

// TODO: FIXME: source should use shared message instead.
int SrsDvr::on_meta_data(SrsOnMetaDataPacket* m) {
    int ret = ERROR_SUCCESS;

    int size = 0;
    char* payload = nullptr;
    if ((ret = m->encode(size, payload)) != ERROR_SUCCESS) {
        return ret;
    }

    SrsSharedPtrMessage metadata;
    if ((ret = metadata.create(nullptr, payload, size)) != ERROR_SUCCESS) {
        return ret;
    }
    if ((ret = plan->on_meta_data(&metadata)) != ERROR_SUCCESS) {
        return ret;
    }
    return ret;
}

int SrsDvr::on_audio(SrsSharedPtrMessage* shared_audio) {
    return plan->on_audio(shared_audio);
}

int SrsDvr::on_video(SrsSharedPtrMessage* shared_video) {
    return plan->on_video(shared_video);
}

///////////////////////////////////////////////////////////////////////////////////////////////

SrsDvrPostFile::SrsDvrPostFile() {
    _stop = true;
}

SrsDvrPostFile::~SrsDvrPostFile() {
    stop();
    if (_thread.joinable() == true) {
        _thread.join();
    }
}

void SrsDvrPostFile::add(const std::string& url, std::vector<std::string>& files) {
    _mutex.lock();

    postRequest req{url, files};
    _list.push_back(std::move(req));

    _mutex.unlock();
}

void dvrPostFile(SrsDvrPostFile* pinfo) {
    if (pinfo == nullptr) {
        return;
    }
    pinfo->run();
}

void SrsDvrPostFile::start() {
    if (_stop == true) {
        _stop = false;
        _thread = std::thread(dvrPostFile, this);
        _thread.detach();
    }
}

void SrsDvrPostFile::stop() {
    _stop = true;
}

void SrsDvrPostFile::run() {
    std::list<postRequest> reqs;
    while (_stop == false) {
        if (_list.empty() == true) {
            //std::this_thread::yield();
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        if (_list.empty() == false) {
            _mutex.lock();
            reqs = std::move(_list);
            _mutex.unlock();
        } else {
            continue;
        }
        if (reqs.empty() == false) {
            for (auto& req : reqs) {
                for (auto& file : req.files) {
                    auto resp = libcurlpfilepath(req.url, srs_path_basename(file), file);
                    if (resp.empty() == true) {
                        continue;
                    }
                    srs_info("post dvr file: %s, to: %s, resp: %s", file.c_str(), req.url.c_str(), resp.c_str());
                    srs_file_remove(file);
                }
            }
            reqs.clear();
        }
    }
}

///////////////////////////////////////////////////////////////////////////////////////////////

static std::vector<SrsDvrPostFile*> g_dvrpfiles;
void initSrsDvrPostFile(int num) {
    auto pflen = (int)g_dvrpfiles.size();
    if (pflen == num) {
        return;
    } else if (pflen > num) {
        for (auto i = num; i < pflen; i++) {
            auto pf = g_dvrpfiles[i];
            if (pf != nullptr) {
                g_dvrpfiles[i] = nullptr;
                pf->stop();
                delete pf;
            }
        }
        g_dvrpfiles.resize(num);
    } else {
        for (auto i = pflen; i < num; i++) {
            auto pf = new SrsDvrPostFile();
            if (pf != nullptr) {
                g_dvrpfiles.push_back(pf);
                pf->start();
            }
        }
    }
}

SrsDvrPostFile* randSrsDvrPostFile() {
    if (g_dvrpfiles.empty() == true) {
        return nullptr;
    }
    auto index = rand() % (int)g_dvrpfiles.size();
    return g_dvrpfiles.at(index);
}

#endif
